package cn.qu.dockman.agent.util;

import cn.qu.dockman.agent.ws.WebSocket;
import cn.qu.dockman.protocol.command.Command;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import javax.security.auth.DestroyFailedException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by zh on 17/3/9.
 */
public class SendingQueue implements Runnable, InitializingBean, DisposableBean {
    private static final Logger logger = LoggerFactory.getLogger(SendingQueue.class);

    private ArrayBlockingQueue<Command> queue = new ArrayBlockingQueue<>(65535);

    private AtomicBoolean closing = new AtomicBoolean(false);

    private AtomicInteger count = new AtomicInteger(0);

    private WebSocket webSocket;

    @Override
    public void run() {
        if (webSocket == null) {
            logger.error("未找到CommandWebSocket对象");
        } else {
            logger.info("开始消费队列");
            while (true) {
                if (count.get() == 0 && closing.get()) {
                    logger.info("队列已消费完毕，销毁中。。。");
                    Thread.currentThread().interrupt();
                    break;
                } else {
                    Command cmd = queue.poll();
                    if (cmd != null) {
                        webSocket.send(cmd.toString());
                    }
                }
            }
        }
    }

    public <T extends Command> void send(T cmd) {
        if (!closing.get()) {
            queue.add(cmd);
            count.addAndGet(1);
        }
    }

    @Override
    public void destroy() throws DestroyFailedException {
        closing.set(true);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        new Thread(this).start();
    }

    public void setWebSocket(WebSocket webSocket) {
        this.webSocket = webSocket;
    }
}
